iT邦幫忙

2026 iThome 鐵人賽

DAY 11
0
自我挑戰組

AI Agent 從零開始系列 第 11 篇

打造「研究助理 Agent」

  • 分享至 

  • xImage
  •  

將前面討論的 Tool System、多工具協調(Multi-Tool Coordination) 與 ReAct 循環 整合,打造一個具備「資料檢索、文章摘要、知識整理與報告撰寫」能力的 研究助理 Agent(Research Assistant Agent)。


1. 研究助理 Agent 的系統架構

研究助理的核心工作不是一次性給出答案,而是透過「搜尋 $\rightarrow$ 閱讀分析 $\rightarrow$ 筆記存檔 $\rightarrow$ 撰寫報告」的漸進式工作流(Iterative Workflow)來完成深度研究。

                       ┌── Web Search Tool(即時資料檢索)
                       ├── Read Page Tool(讀取特定網頁全文)
[Research Agent] ───┼── Note-taking Tool(研究筆記與知識庫存檔)
                       └── Draft Report Tool(寫入最終 Markdown 報告)


2. 完整程式碼實作

以下採用 Python 實作包含「工具註冊」、「狀態紀錄」與「多輪思考執行」的研究助理 Agent:

import inspect
import json
from typing import Callable, Any, Dict, List
from pydantic import BaseModel, create_model
from openai import OpenAI

# ==========================================
# 1. 基礎設施:Tool & ToolRegistry
# ==========================================
class Tool:
    def __init__(self, name: str, description: str, func: Callable):
        self.name = name
        self.description = description
        self.func = func
        self.args_schema = self._generate_args_schema(func)

    def _generate_args_schema(self, func: Callable) -> type[BaseModel]:
        sig = inspect.signature(func)
        fields = {}
        for param_name, param in sig.parameters.items():
            param_type = Any if param.annotation == inspect.Parameter.empty else param.annotation
            default_val = ... if param.default == inspect.Parameter.empty else param.default
            fields[param_name] = (param_type, default_val)
        return create_model(f"{self.name}Schema", **fields)

    def to_openai_tool(self) -> Dict[str, Any]:
        schema = self.args_schema.model_json_schema()
        properties = schema.get("properties", {})
        for prop in properties.values():
            prop.pop("title", None)
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": properties,
                    "required": schema.get("required", [])
                }
            }
        }

    def execute(self, **kwargs) -> str:
        try:
            validated_args = self.args_schema(**kwargs)
            result = self.func(**validated_args.model_dump())
            return result if isinstance(result, str) else json.dumps(result, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"status": "error", "message": f"工具執行失敗: {str(e)}"}, ensure_ascii=False)

class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, Tool] = {}

    def register(self, description: str, name: str = None):
        def decorator(func: Callable):
            tool_name = name or func.__name__
            self._tools[tool_name] = Tool(name=tool_name, description=description, func=func)
            return func
        return decorator

    def get_openai_tools(self) -> List[Dict[str, Any]]:
        return [tool.to_openai_tool() for tool in self._tools.values()]

    def dispatch(self, tool_name: str, arguments_json: str) -> str:
        tool = self._tools.get(tool_name)
        if not tool:
            return json.dumps({"error": f"找不到工具 '{tool_name}'"})
        try:
            args = json.loads(arguments_json) if isinstance(arguments_json, str) else arguments_json
            return tool.execute(**args)
        except Exception as e:
            return json.dumps({"error": f"解析參數失敗: {str(e)}"})


# ==========================================
# 2. 為研究助理註冊「感知與執行工具」
# ==========================================
registry = ToolRegistry()

# 模擬研究助理的記憶/暫存庫
research_notes = []

@registry.register(description="搜尋網頁獲取最新的產業、技術或新聞資訊。輸入搜尋關鍵字,回傳摘要與網址列表。")
def search_web(query: str) -> list:
    # 模擬搜尋引擎 API (如 Tavily, Serper 或 DuckDuckGo)
    if "Agent" in query or "LLM" in query:
        return [
            {"title": "2026 AI Agent 趨勢報告", "url": "https://example.com/ai-agent-2026", "snippet": "2026年 AI Agent 轉向多模態互動與輕量化邊緣運算,自主決策能力顯著提升。"},
            {"title": "從 Chat 到 Agent 的架構演進", "url": "https://example.com/chat-to-agent", "snippet": "Tool Calling 與環境互動成為 Agent 的標準配備,單純對話模式已無法滿足複雜需求。"}
        ]
    return [{"title": "無相關結果", "url": "", "snippet": "請嘗試更精確的搜尋關鍵字。"}]

@registry.register(description="讀取特定網頁網址的完整文字內容,用於深入閱讀與分析細節。")
def fetch_web_page_content(url: str) -> str:
    # 模擬網頁爬蟲/抓取 API
    if "ai-agent-2026" in url:
        return "【詳細內文】2026年,AI Agent 技術核心集中於三個面向:1. 低延遲的推論速度;2. 具備長期記憶的記憶庫管理;3. 跨系統的複雜工作流接軌 (Workflow Integration)。"
    elif "chat-to-agent" in url:
        return "【詳細內文】傳統 Chat Completion 依賴單次 Input/Output,而 Agent 引進了 ReAct 框架,透過 Thought-Action-Observation 循環自動導航至任務目標。"
    return "網頁內容讀取失敗或 404。"

@registry.register(description="將關鍵的研究發現或數據作為筆記記錄下來,供後續撰寫報告使用。")
def add_research_note(topic: str, note_content: str) -> str:
    research_notes.append({"topic": topic, "content": note_content})
    return f"已成功新增主題為 [{topic}] 的研究筆記,目前共有 {len(research_notes)} 條筆記。"

@registry.register(description="查看當前累積的所有研究筆記。")
def get_all_notes() -> list:
    return research_notes


# ==========================================
# 3. 研究助理 Agent 核心 Engine
# ==========================================
class ResearchAssistantAgent:
    def __init__(self, model_name: str = "gpt-4o-mini"):
        self.client = OpenAI()
        self.model_name = model_name

    def run_research(self, research_topic: str, max_rounds: int = 8):
        system_prompt = (
            "你是一位頂尖的專業研究助理。"
            "你的目標是針對使用者的主題進行深入研究,搜集足夠資訊、做筆記,最後產出一份結構完整的 Markdown 研究報告。\n"
            "工作流程建議:\n"
            "1. 使用 search_web 搜尋相關主題資訊。\n"
            "2. 使用 fetch_web_page_content 深入讀取有價值的網頁內文。\n"
            "3. 使用 add_research_note 隨時記錄重要的研究發現。\n"
            "4. 當蒐集的資訊足夠時,閱讀整理所有筆記,並輸出完整的最終研究報告。"
        )

        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"請幫我針對以下主題進行深入研究並撰寫一份簡報分析:'{research_topic}'"}
        ]

        print(f"🔬 開始執行研究任務: {research_topic}\n" + "="*60)

        for round_num in range(1, max_rounds + 1):
            print(f"\n🧠 [Round {round_num}] 研究助理思考中...")

            response = self.client.chat.completions.create(
                model=self.model_name,
                messages=messages,
                tools=registry.get_openai_tools(),
                tool_choice="auto"
            )

            msg = response.choices[0].message
            messages.append(msg)

            # 若模型不再呼叫任何工具,代表研究完成,輸出最終報告
            if not msg.tool_calls:
                print("\n📄 【最終研究報告】\n")
                print(msg.content)
                return msg.content

            # 執行工具
            for call in msg.tool_calls:
                fn_name = call.function.name
                fn_args = call.function.arguments
                print(f"🛠️ [Action] {fn_name}({fn_args})")

                obs = registry.dispatch(fn_name, fn_args)
                print(f"👀 [Observation] {obs[:120]}..." if len(obs) > 120 else f"👀 [Observation] {obs}")

                messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "name": fn_name,
                    "content": obs
                })


# ==========================================
# 4. 啟動 Agent 執行研究
# ==========================================
if __name__ == "__main__":
    agent = ResearchAssistantAgent(model_name="gpt-4o-mini")
    agent.run_research("2026 年 AI Agent 的發展趨勢與架構演進")


3. Agent 的完整思考與執行軌跡

當你啟動研究任務時,Agent 會自發性地展開以下多輪推演(ReAct Loop):

  1. 第一輪:探索性搜尋: Search Web.
    Agent 發現自己對 2026 AI Agent 的細節缺乏具體資訊,發出 Tool Call 呼叫 search_web(query="2026 AI Agent 發展趨勢")。

Observation: 拿到兩個相關網址與簡短 Snippet。

  1. 第二輪:深入閱讀標的網頁: Fetch Full Page Content.
    Agent 分析搜尋結果後,認為這兩篇文章都有價值,決定深入閱讀內容,發出 Tool Call fetch_web_page_content(url="[https://example.com/ai-agent-2026](https://example.com/ai-agent-2026)")。

Observation: 獲得關於低延遲推論、長期記憶與系統整合的詳細內文。

  1. 第三輪:重點整理與筆記紀錄: Add Research Note.
    Agent 讀完文章後,主動將重要資訊結構化,呼叫 add_research_note(topic="2026 趨勢", note_content="三個核心面向:低延遲推論、記憶管理與系統整合。")。

Observation: "已成功新增主題...目前共有 1 條筆記。"

  1. 第四輪:資訊彙整與成果交付: Draft Final Report.
    Agent 評估資料已足夠回答使用者的問題,終止工具呼叫,直接利用筆記內容撰寫出包含「摘要、核心趨勢、架構演進與未來展望」的結構化 Markdown 報告。

4. 生產級研究助理 Agent 的高階防護與擴充

要將這個專案部署到真實環境,還需補齊以下關鍵模組:

  1. 真實工具串接 (Real Tools Interfacing):
  • 搜尋:替換為 Tavily API 或 Google Custom Search API(專為 LLM 優化的搜尋介面)。
  • 讀頁:串接 Jina Reader 或 Playwright,自動將 HTML 轉為乾淨的 Markdown 格式,避免 Token 浪費。
  1. 上下文壓縮與長文本處理 (Context Pruning):
  • 在 fetch_web_page_content 時,網頁全文可能包含數萬字。可以在 Tool 內部加上「小模型預先摘要(Mini-Summary)」機制,只把精煉後的 500 字摘要塞回 Agent 的上下文。
  1. 結構化報告範本 (Structured Output Enforcement):
  • 若對最終輸出的 Markdown 格式有嚴格要求(如必須包含大綱、參考來源連結表),可以強制要求 Agent 在最後一輪使用 Pydantic 指定格式進行輸出。

上一篇
Tool 不只是 Function
下一篇
Agent Loop——Agent 為什麼會一直工作?
系列文
AI Agent 從零開始 共 25 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言